package com.deng.hadoop.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

/**
 * Created By DengLibin
 * Date: 19-4-1
 * Time: 上午9:22
 * Desc: map-reduce程序 单词计数 客户端
 * 打成jar包后 用hadoop运行 （hadoop jar MyWordCount.jar  /data/wordcount.txt /data/myout）
 */
public class WordCountClient {

    /**
     * 输入参数２个，　输入文件的路径，结果输出文件的路径
     * @param args
     * @throws IOException
     * @throws ClassNotFoundException
     * @throws InterruptedException
     */
    public  static  void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        //获取输入参数
        String[] cliArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (cliArgs.length != 2) {
            System.err.println("Usage: MyWordCount <in> <out>");
            System.exit(2);
        }
        //创建作业
        Job myJob = Job.getInstance(conf, "MyWordCount");
        //作业类
        myJob.setJarByClass(WordCountClient.class);
        //mapper类
        myJob.setMapperClass(WordCountMapper.class);
        //设置map的输出的key-value类型
        myJob.setMapOutputKeyClass(Text.class);
        myJob.setMapOutputValueClass(IntWritable.class);
        //reducer类
        myJob.setReducerClass(WordCountReducer.class);
        //设置Combiner类（作用：在一个map内做一次组合，减少传输到reduce端的数据量,）
        /**
         *
           每一个map可能会产生大量的输出，combiner的作用就是在map端对输出先做一次合并，以减少传输到reducer的数据量。
         combiner最基本是实现本地key的归并，combiner具有类似本地的reduce功能。 如果不用combiner，那么，所有的结果
         都是reduce完成，效率会相对低下。使用combiner，先完成的map会在本地聚合，提升速度。
         注意：Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以Combiner只应该
         　　用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。
         * */
        myJob.setCombinerClass(WordCountReducer.class);

        //输入（来自文件，也可以来自数据库，文本等，可以加多个）
        FileInputFormat.addInputPath(myJob, new Path(cliArgs[0]));
        Path outPath = new Path(cliArgs[1]);
        //如果存在就递归删除
        if(outPath.getFileSystem(conf).exists(outPath)){
            outPath.getFileSystem(conf).delete(outPath, true);
        }
        //输出（到文件，也可以到数据库）
        FileOutputFormat.setOutputPath(myJob, outPath);
        //设置切片(Split)大小默认和块大小一致
        //FileInputFormat.setMinInputSplitSize(myJob, 128L * 1024L * 1024L);
        boolean success = myJob.waitForCompletion(true);
        System.out.println("result:" + success);
        System.exit(success ? 0: 1);
    }
}
